package com.runa.monitor.platinterface.mqtt;

import com.alibaba.fastjson.JSON;
import com.runa.monitor.comm.StormAIData.*;
import com.runa.monitor.comm.constants.ResultConstants;
import com.runa.monitor.comm.dto.newComm.*;
import com.runa.monitor.comm.entity.CommMeter;
import com.runa.monitor.comm.entity.CommNetEquipment;
import com.runa.monitor.comm.entity.CommPanel;
import com.runa.monitor.comm.entity.CommValve;
import com.runa.monitor.comm.enums.NetFaultCode;
import com.runa.monitor.platinterface.base.AnalysisFactory;
import com.runa.monitor.platinterface.base.BaseContainer;
import com.runa.monitor.platinterface.service.GwService;
import com.runa.persistence.util.LogUtil;
import com.runa.persistence.util.SpringContextUtil;
import com.runa.monitor.platinterface.extend.nb.vo.SendCmdParamVo;
import com.runa.monitor.platinterface.netservice.equipment.NetEqu;
import com.runa.monitor.platinterface.netservice.equipment.RunaLDtuDriver;
import com.runa.monitor.platinterface.platService.NetEquService;
import com.runa.monitor.platinterface.platService.RepositoryService;
import com.runa.monitor.platinterface.platService.gateway.BuildGatewayService;
import com.runa.monitor.platinterface.platService.gateway.GatewayMessageTrans;
import com.runa.persistence.DateUtil;
import com.runa.persistence.SerializeUtil;
import com.runa.persistence.repository.CommMeterRepository;
import com.runa.protocol.analysis.ShellBufferAnalysis;
import com.runa.protocol.util.ByteUtil;
import com.runa.protocol.util.UtilTime;
import com.runa.rccprotocol.analysis.RunaDeviceAnalysis;
import com.runa.rccprotocol.dto.ConcentratorResponseData;
import com.runa.rccprotocol.dto.equdata.ConcenReportData;
import lombok.extern.slf4j.Slf4j;
import org.assertj.core.util.Lists;
import org.springframework.data.domain.Example;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

import static com.runa.monitor.platinterface.extend.nb.service.NbPlantService.getSendCmdByParamList;

@Component
@EnableAsync
@Slf4j
public class MQTTExecutor {

    @Resource
    private RUEquService ruEquService;

    @Resource
    private CommMeterRepository commMeterRepository;

    @Resource
    private RepositoryService repositoryService;
    @Resource
    private GwService gwService;

    public static void main(String[] args) {
        System.out.println(UtilTime.StringToDate("0000-00-00 00:00:00", "yyyy-MM-dd HH:mm:ss").getTime());
    }

    @Async("mqttResponseScheduler")
    public void sendLDTU(String up, byte[] data) {
        NetEquService netEquService = SpringContextUtil.getBean(NetEquService.class);
        if (up != null) {
            String[] kvs = up.split(",");
            Map<String, String> param = Lists.list(kvs)
                    .stream()
                    .collect(Collectors.toMap(kv -> kv.substring(0, kv.indexOf(":")),
                            kv -> kv.substring(kv.indexOf(":")+1)));
            String netSerial = param.get("id");
            netEquService.removeInstance(netSerial);
            NetEqu<String> netEqu = netEquService.getNetEquBySerial(netSerial);
            ConcenReportData reportData = new ConcenReportData();
            reportData.setSerial(param.get("id"));
            reportData.setCsq(Integer.valueOf(param.get("csq")));
            reportData.setDate(new Date());
            reportData.setIccid(param.get("icc"));
//            reportData.setElectric(Float.parseFloat(param.get("bat")));
            reportData.setVer(param.get("ver"));
            BaseContainer.netStatusCallback().rccNetEquReport(param.get("id"), reportData);
            if (netEqu instanceof RunaLDtuDriver) {
                log.info("LDTU:{} 执行任务", netEqu.serial());
                netEqu.heartbeatUpdate();
                ((RunaLDtuDriver) netEqu).sentDateTime();
                ((RunaLDtuDriver) netEqu).sentBuildValves();
                ((RunaLDtuDriver) netEqu).readBuildValves();
            }
        } else if (data != null) {
            ShellBufferAnalysis<byte[]> analysis = AnalysisFactory.create("RUNA_NATIVE");
            if (analysis != null) {
                analysis.fill(data, data.length, null);
                ConcentratorResponseData responseData = (ConcentratorResponseData) analysis.gain();
                NetEqu netEqu = netEquService.getNetEquBySerial(responseData.getAddress());
                if (netEqu != null) {
                    netEqu.receive(responseData);
                }
            }
        }
    }

    @Async("mqttResponseScheduler")
    public void sendSms(String topic, String json) {
        try {
            if (json.startsWith("func:up,")) {
                ruEquService.execute(topic, json);
                return;
            } else if (json.startsWith("FE")) {
                ruEquService.executeModbus(topic, json);
                return;
            } else if (topic.startsWith("/RU-METER") || topic.startsWith("/HM-1")) {
                ruEquService.executeMeter(topic, json);
            }
            if (json.contains("\"Func\":\"up\"")) {
                UpstreamMessage message = SerializeUtil.json2pojo(json, UpstreamMessage.class);
                if (message != null) {
                    AsynResult asynResult = new AsynResult();
//                    String model = MqttTool.getModelFromTopic(topic);
                    CommNetEquipment commNetEquipment;
//                    if (StringUtils.isEmpty(model)) {
                        commNetEquipment = repositoryService.getCommNetEquipmentByNetSerial(message.getId());
//                    } else {
//                        commNetEquipment = repositoryService.getCommNetEquipmentByModelAndSerial(model, message.getId());
//                    }
                    if (commNetEquipment == null) {
                        log.error("无此档案---：网关编号:" + message.getId());
                        return;
                    }
                    asynResult.setIdentity(commNetEquipment.getId());
                    asynResult.setCode(ResultConstants.CODE_200);
                    asynResult.setMsg(MqttUtil.successString);
                    if ("gwlogin".equals(message.getParam().getType())) {//网关上线
                        log.debug("收到网关客户端上线消息，{}", json);
                        BaseContainer.netStatusCallback().rccStatusChange(message.getId(), true);
                        gwService.downTiming(message.getId());
                    } else if ("event".equals(message.getParam().getType())) {//事件上报
                        log.debug("收到mqtt客户端楼网关事件上报，{}", json);
                        if ((message.getParam().getData().toString()).contains("powerdown")) {
                            NetCurrentData netCurrentData = new NetCurrentData();
                            netCurrentData.setEvent("powerdown");
                            netCurrentData.setDate(new Date());
                            asynResult.setContent(netCurrentData);
                            MqttUtil.send(NetCurrentData.TOPIC, asynResult);
                        }
                    } else {
                        if ("devicedata".equals(message.getParam().getType())) {
                            List<Map<String, Object>> dataArray = (List<Map<String, Object>>) message.getParam().getData();
                            if (CollectionUtils.isEmpty(dataArray)) {
                                log.debug("当前网关无数据，消息体为空,{}", json);
                                return;
                            }
                            for (Map<String, Object> data : dataArray) {
                                String deviceType = (String) data.get("Type");
                                Map<String, Object> map = (Map<String, Object>) data.get("Data");
                                dataGen(asynResult, deviceType, map, commNetEquipment);
                            }
                        } else if ("gwdata".equals(message.getParam().getType())) {
                            NetCurrentData netCurrentData = new NetCurrentData();
                            Map<String, String> data = (Map<String, String>) message.getParam().getData();
                            netCurrentData.setDate(DateUtil.toDateGeneric(data.get("Now")));
                            if (!StringUtils.isEmpty(data.get("ModemCsq"))) {
                                netCurrentData.setSignValue(Integer.parseInt(data.get("ModemCsq").trim()));
                            }
                            if (!StringUtils.isEmpty(data.get("Softver"))) {
                                netCurrentData.setVer(data.get("Softver"));
                            }
                            if (!StringUtils.isEmpty(data.get("Hardver"))) {//硬件版本
                                netCurrentData.setVer(netCurrentData.getVer()+"-"+data.get("Hardver"));
                            }
                            if (!StringUtils.isEmpty(data.get("ModemIccid"))) {
                                netCurrentData.setIccid(data.get("ModemIccid"));
                            }
                            if (!StringUtils.isEmpty(data.get("EtrVal")))
                                netCurrentData.setElectric(Float.valueOf(data.get("EtrVal")));
                            if (!StringUtils.isEmpty(data.get("EtrComm")))
                                netCurrentData.setElectricComm(data.get("EtrComm"));
                            if (!StringUtils.isEmpty(data.get("EtrTime")))
                                netCurrentData.setElectricTime(DateUtil.toDateGeneric(data.get("EtrTime")));
                            asynResult.setContent(netCurrentData);
//                            asynResult.setIdentity(repositoryService.getCommNetEquipmentByNetSerial(asynResult.getIdentity()).getId());
                            MqttUtil.send(NetCurrentData.TOPIC, asynResult);
                        }
                    }
                }
            } else if (json.contains("\"down_resp\"")) {
                DownRespMessage message = SerializeUtil.json2pojo(json, DownRespMessage.class);
                AsynResult asynResult = new AsynResult();
                CommNetEquipment commNetEquipment = repositoryService.getCommNetEquipmentByNetSerial(message.getId());
                if (commNetEquipment == null) {
                    log.error("无此档案---：网关编号:" + message.getId());
                    return;
                }
                asynResult.setIdentity(commNetEquipment.getId());
                //先行停掉超时部分
                BuildGatewayService buildGatewayService = GatewayMessageTrans.buildGatewayServiceHashMap.get(message.getSeq());
                if (buildGatewayService == null) {
                    log.error("该任务不存在: {}", message.getSeq());
                    return;
                }
                GatewayMessageTrans.buildGatewayServiceHashMap.remove(message.getSeq());
                buildGatewayService.mockUnlock();
                if (message.getSeq().contains("-")) {
                    String[] split = message.getSeq().split("-");
                    asynResult.setTask(split[0]);
                    asynResult.setHoldInfo(split[1]);
                }
                String s = message.getSeq().split("-")[2];
                if (message.getResult() instanceof String) {
                    if (!"00000000".equalsIgnoreCase(s)) asynResult.setIdentity(s);
                    if ("ok".equals(message.getResult())) {
                        asynResult.setCode(ResultConstants.CODE_200);
                        asynResult.setMsg(MqttUtil.successString);
                    } else {
                        asynResult.gatewayReturnFailureSetting();
                    }
                    MqttUtil.send(MqttUtil.topicOfNoneContent, asynResult);
                } else if (message.getResult() instanceof Map) {//对应节点设备回应
                    Map<String, Object> result = (Map<String, Object>) message.getResult();
                    Map<String, Object> data = (Map<String, Object>) result.get("Data");
                    if ("rddevice".equals(result.get("Type"))) {
                        String deviceType = (String) data.get("Type");
                        if ((boolean) data.get("Com")) {
                            asynResult.setCode(ResultConstants.CODE_200);
                            asynResult.setMsg(MqttUtil.successString);
                            Map<String, Object> dataData = (Map<String, Object>) data.get("Data");
                            dataGen(asynResult, deviceType, dataData, commNetEquipment);
                        } else {
                            asynResult.unreadSetting();
                            if (!"00000000".equalsIgnoreCase(s)) asynResult.setIdentity(s);
                            switch (deviceType) {
                                case "housemeter":
                                case "buildmeter":
                                    MqttUtil.send(HotMeterVO.TOPIC, asynResult);
                                    break;
                                case "buildvalve":
                                case "housevalve":
                                    MqttUtil.send(ValveVO.TOPIC, asynResult);
                                    break;
                                case "temp":
                                    MqttUtil.send(PanelVO.TOPIC, asynResult);
                                    break;
                            }
                        }
                    } else if ("getparam".equals(result.get("Type"))) {
                        NetCurrentData netCurrentData = new NetCurrentData();
                        netCurrentData.setMinuteOfCollectForHouse(Integer.valueOf(data.get("house_collect_interval").toString()));
                        netCurrentData.setMinuteOfUploadForHouse(Integer.valueOf(data.get("house_upload_interval").toString()));
                        netCurrentData.setMinuteOfCollectForBuild(Integer.valueOf(data.get("build_collect_interval").toString()));
                        netCurrentData.setMinuteOfUploadForBuild(Integer.valueOf(data.get("build_upload_interval").toString()));
                        netCurrentData.setDate(new Date());
                        asynResult.setContent(netCurrentData);
                        asynResult.setCode(ResultConstants.CODE_200);
                        asynResult.setMsg(MqttUtil.successString);
                        MqttUtil.send(NetCurrentData.TOPIC, asynResult);
                    }
                }
            } else if (json.contains("clientid")) {//客户端掉线
                try {
                    Map message = SerializeUtil.json2map(json);
                    assert message != null;
                    String clientid = message.get("clientid").toString();
                    if (clientid.contains("aigw_")) {
                        String serial = "";
                        if (clientid.startsWith("aigw_jy_")) serial = clientid.replace("aigw_jy_", "");
                        if (clientid.startsWith("aigw_mt_")) serial = clientid.replace("aigw_mt_", "");
                        if (clientid.startsWith("aigw_")) serial = clientid.replace("aigw_", "");
                        CommNetEquipment commNetEquipment = repositoryService.getCommNetEquipmentByNetSerial(serial);
                        if (commNetEquipment == null) {
                            log.error("数据库无此网关档案" + serial);
                            return;
                        }
                        BaseContainer.netStatusCallback().rccStatusChange(serial, false);
                    }
                } catch (Exception e) {
                    log.error("网关掉线处理失败 ", e);
                }
            }
        } catch (Exception e) {
            log.error("Build Publish Consumer Accept: ", e);
        }
    }


    private void dataGen(AsynResult asynResult, String deviceType, Map<String, Object> map, CommNetEquipment commNetEquipment) {
        asynResult.setTimestamp(System.currentTimeMillis());
        AsynResult asynResultT;
        switch (deviceType) {
            case "housemeter":
            case "buildmeter":
                HotMeterVO hotMeterVO = meterCurrentData(map);
                CommMeter meter = repositoryService.getCommMeterByNodeSerialForNetEqu(hotMeterVO.getMeterSerial(), asynResult.getIdentity());
                if (meter == null) {
                    log.error("meter此编号无此档案：" + hotMeterVO.getMeterSerial() + "---" + asynResult.getIdentity());
                    return;
                }
                asynResultT = JSON.parseObject(JSON.toJSONString(asynResult), AsynResult.class);
                asynResultT.setIdentity(meter.getId());
                if (hotMeterVO.getFaultCode() != null && NetFaultCode.DEVICE_NOT_READ.getName().equalsIgnoreCase(hotMeterVO.getFaultCode().get(0))) {
                    asynResultT.unreadSetting();
                } else {
                    asynResultT.setContent(hotMeterVO);
                }
                LogUtil.info(commNetEquipment.getSerial(), JSON.toJSONString(asynResultT));
                MqttUtil.send(HotMeterVO.TOPIC, asynResultT);
                break;
            case "buildvalve":
                ValveVO valveVO = lzValveCurrentData(map);
                CommValve valve = repositoryService.getCommValveByNodeSerialForNetEqu(valveVO.getValveSerial(), asynResult.getIdentity());
                if (valve == null) {
                    log.error("buildvalve此编号无此档案：" + valveVO.getValveSerial());
                    return;
                }
                asynResultT = JSON.parseObject(JSON.toJSONString(asynResult), AsynResult.class);
                asynResultT.setIdentity(valve.getId());
                if (valveVO.getFaultCode() != null && NetFaultCode.DEVICE_NOT_READ.getName().equalsIgnoreCase(valveVO.getFaultCode().get(0))) {
                    asynResultT.unreadSetting();
                } else {
                    asynResultT.setContent(valveVO);
                }
                MqttUtil.send(ValveVO.TOPIC, asynResultT);
                break;
            case "housevalve":
                List<CurrentData> currentData = valveCurrentData(map);
                ValveVO valveT = (ValveVO) currentData.get(0);
                CommValve commValve = repositoryService.getCommValveByNodeSerialForNetEqu(valveT.getValveSerial(), asynResult.getIdentity());
                if (commValve == null) {
                    log.error("housevalve此编号无此档案：" + valveT.getValveSerial());
                    return;
                }
                asynResultT = JSON.parseObject(JSON.toJSONString(asynResult), AsynResult.class);
                asynResultT.setIdentity(commValve.getId());
                if (valveT.getFaultCode() != null && NetFaultCode.DEVICE_NOT_READ.getName().equalsIgnoreCase(valveT.getFaultCode().get(0))) {
                    asynResultT.unreadSetting();
                } else {
                    asynResultT.setContent(valveT);
                }
                MqttUtil.send(ValveVO.TOPIC, asynResultT);
                if (currentData.size() > 1) {//面板
                    PanelVO panel = (PanelVO) currentData.get(1);
                    CommPanel commPanel = repositoryService.getCommPanelByNodeSerialForNetEqu(panel.getSerial(), asynResult.getIdentity());
                    if (commPanel == null) {
                        log.error("housevalve-panel此编号无此档案：" + panel.getSerial());
                        return;
                    }
                    asynResultT = JSON.parseObject(JSON.toJSONString(asynResult), AsynResult.class);
                    asynResultT.setIdentity(commPanel.getId());
                    if (panel.getFaultCode() != null && ("抄读失败".equals(panel.getFaultCode().get(0)) || NetFaultCode.DEVICE_NOT_READ.getName().equalsIgnoreCase(panel.getFaultCode().get(0)))) {
                        log.info("户阀面板抄读失败: {} {}", panel.getSerial(), panel.getFaultCode().get(0));
                        asynResultT.unreadSetting();
                    } else {
                        asynResultT.setContent(Collections.singleton(panel));
                    }
                    MqttUtil.send(PanelVO.TOPIC, asynResultT);
                }
                break;
            case "housetemp":
                PanelVO panelVO = panelCurrentData(map);
                CommPanel commPanel = repositoryService.getCommPanelByNodeSerialForNetEqu(panelVO.getSerial(), asynResult.getIdentity());
                if (commPanel == null) {
                    log.error("housetemp此编号无此档案：" + panelVO.getSerial());
                    return;
                }
                asynResultT = JSON.parseObject(JSON.toJSONString(asynResult), AsynResult.class);
                asynResultT.setIdentity(commPanel.getId());
                if (panelVO.getFaultCode() != null && ("抄读失败".equals(panelVO.getFaultCode().get(0)) || NetFaultCode.DEVICE_NOT_READ.getName().equalsIgnoreCase(panelVO.getFaultCode().get(0)))) {
                    log.info("户阀面板上报失败: {} {}", panelVO.getSerial(), panelVO.getFaultCode().get(0));
                    asynResultT.unreadSetting();
                } else {
                    asynResultT.setContent(Collections.singleton(panelVO));
                }
                MqttUtil.send(PanelVO.TOPIC, asynResultT);
                break;
        }
    }

    private HotMeterVO meterCurrentData(Map<String, Object> data) {
        HotMeterVO currentData = new HotMeterVO();
        MeterData meterData = SerializeUtil.map2pojo(data, MeterData.class);
        currentData.setMeterSerial(meterData.getId());
        if (!StringUtils.isEmpty(meterData.getStatus()) && (meterData.getStatus().equals("抄读失败"))) {
            currentData.setFaultCode(Collections.singletonList(NetFaultCode.DEVICE_NOT_READ.getName()));
            currentData.setStatus(meterData.getStaword());
            return currentData;
        }
        currentData.setReadDate(meterData.getReadTime());
        if (meterData.getDateTime() != null
                && meterData.getDateTime().getTime() > 0) {
            currentData.setMeterDate(meterData.getDateTime());
        }
        currentData.setFlow(meterData.getLsl());
        currentData.setPower(meterData.getRgl());
        currentData.setAccFlow(meterData.getLl());
        currentData.setAccHeat(meterData.getRl());
        currentData.setSupplyTemp(meterData.getSupTemp());
        currentData.setReturnTemp(meterData.getRetTemp());
        currentData.setWorkTime(meterData.getWorkTime());
        currentData.setSuccess(true);
        currentData.setStatus(meterData.getStaword());
        currentData.setFaultCode(meterData.getStatus());
        currentData.setSignInfo(meterData.getRssi());//信号强度
        return currentData;
    }

    private ValveVO lzValveCurrentData(Map<String, Object> data) {
        ValveVO currentData = new ValveVO();
        BuildValveData valveData = SerializeUtil.map2pojo(data, BuildValveData.class);
        currentData.setValveSerial(valveData.getId());
        if (!StringUtils.isEmpty(valveData.getStatus()) && (valveData.getStatus().equals("抄读失败"))) {
            currentData.setFaultCode(NetFaultCode.DEVICE_NOT_READ.getName());
            currentData.setStatus(valveData.getStaword());
            return currentData;
        }
        currentData.setSuccess(true);
        currentData.setReadTime(valveData.getReadTime());
        currentData.setLR(valveData.getMode() == 1 ? ValveVO.LREnum.本地 : ValveVO.LREnum.远程);
//        currentData.setControlObj(valveData.getTactics());
        currentData.setRemoteSet((float) valveData.getSetDeg());
        currentData.setOuputOpen((float) valveData.getSetDeg());
        int realDeg = valveData.getRealDeg() == 255 ? -1 : valveData.getRealDeg() == 254 ? -2 :
                valveData.getRealDeg() == 101 ? 100 : valveData.getRealDeg();
        currentData.setActOpen((float) realDeg);
        currentData.setSupplyTemp(valveData.getSupTemp());
        currentData.setReturnTemp(valveData.getRetTemp());
        currentData.setSupplyPre(valveData.getSupPres());
        currentData.setBehindPre(valveData.getRetPresAfter());
        currentData.setFrontPre(valveData.getRetPresBefor());
        currentData.setLR(valveData.getMode() == 1 ? ValveVO.LREnum.远程 : ValveVO.LREnum.本地);
        currentData.setSoftVersion(String.valueOf(valveData.getVersion()));
        currentData.setStatus(valveData.getStaword());//状态字
        currentData.setSignal(valveData.getRssi());//信号强度
        currentData.setBatteryVol(valveData.getBatteryVoltage());
        currentData.setFaultCode(valveData.getStatus());
        return currentData;
    }

    private List<CurrentData> valveCurrentData(Map<String, Object> data) {
        List<CurrentData> currentDataList = new ArrayList<>();
        ValveVO currentData = new ValveVO();
        HouseValveData valveData = SerializeUtil.map2pojo(data, HouseValveData.class);
        currentData.setValveSerial(valveData.getId());
        if (!StringUtils.isEmpty(valveData.getStatus()) && (valveData.getStatus().equals("抄读失败"))) {
            currentData.setFaultCode(NetFaultCode.DEVICE_NOT_READ.getName());
            currentData.setStatus(valveData.getStaword());
            currentDataList.add(currentData);
            return currentDataList;
        }
        currentData.setReadTime(valveData.getReadTime());
        if (valveData.getDateTime() != null
                && valveData.getDateTime().getTime() > 0) {
            currentData.setValveDate(valveData.getDateTime());
        }
        currentData.setOpenTime((float) valveData.getKf());
        //楼网关定义不明，暂且全部以开度为准
        currentData.setModel(ValveVO.ModelEnum.开度);
        currentData.setRemoteSet(valveData.getVlckDst());
        currentData.setTempLock(valveData.getTlckFlag() == 1);
        currentData.setTempUL(valveData.getTemphLimit());
        currentData.setTempDL(valveData.getTemplLimit());
        currentData.setSupplyTemp(valveData.getSupTemp());
        currentData.setReturnTemp(valveData.getRetTemp());
        currentData.setOpenLock(valveData.getVlckFlag() == 1);
        int realDeg = valveData.getRealDeg() == 255 ? -1 : valveData.getRealDeg() == 254 ? -2 :
                valveData.getRealDeg() == 101 ? 100 : valveData.getRealDeg();
        currentData.setActOpen((float) realDeg);
        currentData.setStatus(valveData.getStaword());
        currentData.setFaultCode(valveData.getStatus());
        currentData.setSignal(valveData.getRssi());//信号强度
        currentDataList.add(currentData);
        HouseTempData houseTempData = valveData.getHouseTempData();
        if (houseTempData != null) {
            PanelVO panelCurrentData = new PanelVO();
            panelCurrentData.setSerial(houseTempData.getId());
            panelCurrentData.setSuccess(true);
            panelCurrentData.setReadTime(houseTempData.getReadTime());
            panelCurrentData.setPanelDate(houseTempData.getReadTime());
            panelCurrentData.setCounter(houseTempData.getTpCnt());
            panelCurrentData.setStatus(houseTempData.getTpStateword());
            //TODO 需要区分温控还是测温
            panelCurrentData.setPanelType("温控");
            panelCurrentData.setFaultCode(houseTempData.getStatus());
            panelCurrentData.setTemperature(houseTempData.getCurTemp());
//            panelCurrentData.(houseTempData.getRssi());//信号强度  实际无
            currentDataList.add(panelCurrentData);
        }
        return currentDataList;
    }

    private PanelVO panelCurrentData(Map<String, Object> data) {
        PanelVO currentData = new PanelVO();
        HouseTempData tempData = SerializeUtil.map2pojo(data, HouseTempData.class);
        currentData.setSerial(tempData.getId());
        currentData.setSuccess(true);
        currentData.setReadTime(tempData.getReadTime());
        if (tempData.getDateTime() != null
                && tempData.getDateTime().getTime() > 0) {
            currentData.setPanelDate(tempData.getDateTime());
        }
        currentData.setCounter(tempData.getTpCnt());
        currentData.setStatus(tempData.getTpStateword());
        currentData.setPanelType("温控");
        currentData.setFaultCode(tempData.getStatus());
        currentData.setTemperature(tempData.getCurTemp());
//        currentData.setSignInfo(tempData.getRssi());//信号强度
        return currentData;
    }

    @Async("mqttResponseScheduler")
    public void sendHotmeter(String msg) {
        if (msg != null) {
            String[] kvs = msg.split(",");
            Map<String, String> param = Lists.list(kvs)
                    .stream()
                    .collect(Collectors.toMap(kv -> kv.substring(0, kv.indexOf(":")), kv -> kv.substring(kv.indexOf(":")+1)));
            String serial = param.get("id");
            CommMeter commMeter = commMeterRepository.findOne(Example.of(CommMeter.builder().serial(serial)
                    .build())).orElse(null);
            if (commMeter == null) {
                log.error("数据库无数据---" + JSON.toJSONString(msg));
                return;
            }
            String data = param.get("data");
            data = data.replace("end.", "");
            RunaDeviceAnalysis runaDeviceAnalysis = new RunaDeviceAnalysis();
            HotMeterVO hotMeterVO = runaDeviceAnalysis.heatMeterData(9, ByteUtil.hexStringToByte(data));
            hotMeterVO.setMeterSerial(param.get("id"));
            hotMeterVO.setSignInfo((param.get("csq")));
            hotMeterVO.setReadDate(new Date());
            hotMeterVO.setMeterDate(new Date());
            hotMeterVO.setBatteryVol(Float.parseFloat(param.get("volt")));
//            hotMeterVO.set(param.get("sw"));
            AsynResult asynResult = AsynResult.success();
            asynResult.setTimestamp(new Date().getTime());
            asynResult.setIdentity(commMeter.getId());
            asynResult.setContent(hotMeterVO);
            MqttUtil.send(HotMeterVO.TOPIC, asynResult);

            //下发校时
            List<SendCmdParamVo> sendCmdParamVoList = new ArrayList<>();
            sendCmdParamVoList.add(new SendCmdParamVo("rtc:", UtilTime.getCurrentTime("yyyyMMddHHmmss")));
            String sendContent = getSendCmdByParamList(sendCmdParamVoList);
            sendContent = sendContent + "\r\n";
            log.info("cat1 阀---下发消息：" + hotMeterVO.getMeterSerial() + "---" + JSON.toJSONString(sendContent));
            MqttBaseController.getConnector().sendMessage("/hotmeter-00/" + hotMeterVO.getMeterSerial() + "/down", sendContent);
        }
    }
}

